package com.lpf;

import org.apache.http.HttpHost;
import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;

import java.io.IOException;

/**
 * 批量操作 + 监听
 */
public class Main09 {

    public static void main(String[] args) {

        RestHighLevelClient client = new RestHighLevelClient(
                RestClient.builder(
                        new HttpHost("localhost", 9200, "http"),
                        new HttpHost("localhost", 9200, "http")
                )
        );

        // 该 Listener 可以监听每个 BulkRequest 请求相关事件
        BulkProcessor.Listener listener = new BulkProcessor.Listener() {
            public void beforeBulk(long executionId, BulkRequest request) {
                System.out.println("beforeBulk"); // 在 BulkRequest 请求发送之前，会触发该方法
            }

            public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                System.out.println("afterBulk"); // 在收到 BulkResponse响应时触发该方法，这里可以通过response.hasFailures()方法判断请求是否失败
            }

            public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                System.out.println("afterBulk2"); // 在 BulkRequest请求抛出异常的时候，会触发该方法
            }
        };

        // 创建 BulkProcessor
        BulkProcessor bulkProcessor = BulkProcessor
                .builder(client::bulkAsync, listener)
                .setBulkActions(10)
                .setBulkSize(new ByteSizeValue(5, ByteSizeUnit.MB))
                .setFlushInterval(TimeValue.timeValueSeconds(10))
                .setConcurrentRequests(1)
                .setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100), 3))
                .build();

        // 创建 IndexRequest 请求，这里需要指定 Index 名称
        IndexRequest request01 = new IndexRequest("skywalking");
        request01.type("type");
        request01.id("2");  // Document Id，不指定的话，ElasticSearch 为其自动分配

        // Document的具体内容
        String jsonString = "{\"user\":\"kim\",\"postDate\":\"2013-01-30\",\"age\":29,\"message\":\"trying out Elasticsearch\"}";
        request01.source(jsonString, XContentType.JSON);

        bulkProcessor.add(request01);

        DeleteRequest request02 = new DeleteRequest("skywalking", "type", "1");
        bulkProcessor.add(request02);
    }
}